package cn.doitedu.dwetl.etl

import java.text.{DateFormat, SimpleDateFormat}
import java.util.Date

import ch.hsr.geohash.GeoHash
import cn.doitedu.commons.utils.SparkUtil
import cn.doitedu.dwetl.beans.EventLogBean
import cn.doitedu.dwetl.utls.{BeanUtil, BindTableUtil}
import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.lionsoul.ip2region.{DbConfig, DbSearcher}

/**
 * @Author hunter@doitedu.cn
 * @date 2020/8/215:45
 * @Description: ODS -> DWD 的etl程序
 *    要需求：1. json解析，清洗，过滤
 *    2.地理位置解析（先解析gps，再用ip解析）
 *    3.GUID的标识
 *    4.新老用户的标识
 *
 *   用到的数据：1.当天的事件日志
 *   2.前一日 和今日的 idmapping 绑定表
 *   3.geohash地理位置表
 *   3.ip地理位置表
 *
 */
object Ods2Dwd {

  def main(args: Array[String]): Unit = {
    var cur_dt = ""
    var pre_dt = ""
    var log_y = ""
    var log_m = ""
    var log_d = ""

    try {
      cur_dt = args(0)
      pre_dt = args(1)

      log_y = cur_dt.split("-")(0)
      log_m = cur_dt.split("-")(1)
      log_d = cur_dt.split("-")(2)
    } catch {
      case e: Exception => println(
        """
          |
          |Usage:
          |参数1：要处理的日志数据的日期,如:2020-07-31
          |参数2：前一天设备绑定表日期,如:2020-07-30
          |参数3：程序要运行的模式：local|yarn
          |
          |""".stripMargin)
        sys.exit(1)
    }
    val spark = SparkUtil.getSparkSession("ODS2DWD数据处理",args(2))
    import spark.implicits._

    // 1. 从hive的ods库的app_event_log表加载当天日志数据
    val eventLog = spark
      .read
      .table("ods.app_event_log")
      .where(s"y=${log_y} and m=${log_m}  and d=${log_d}")

    // 2. 脏数据过滤， 关键字段缺失，时间段是否符合，爬虫请求  等过滤
    val washedEventLog = eventLog.rdd.map(BeanUtil.row2EventLogBean(_))
      .filter(bean => {
        var flag = true
        if (bean == null) {
          flag = false
        } else {
          if (StringUtils.isBlank(bean.account) && StringUtils.isBlank(bean.deviceid)) flag = false
          if (StringUtils.isBlank(bean.eventid) || StringUtils.isBlank(bean.sessionid) || bean.properties == null) flag = false
        }

        val sdf = new SimpleDateFormat("yyyy-MM-dd")

        val startTime = sdf.parse(s"${cur_dt}").getTime
        val endTime = startTime+24*60*60*1000

        if (bean.timestamp > endTime || bean.timestamp < startTime) flag = false

        flag
      })

    // 3. 地理位置解析
    // 加载geo地理位置字典，并广播
    val geoDF = spark.read.table("dim.geo_dict")
    val geoMap = geoDF.rdd.map(row => {
      val geo = row.getAs[String]("geo")
      val province = row.getAs[String]("province")
      val city = row.getAs[String]("city")
      val district = row.getAs[String]("district")
      (geo, (province, city, district))
    }).collectAsMap()
    val bc = spark.sparkContext.broadcast(geoMap)

    // 从hdfs中加载ip地址库数据，并广播
    val conf = new Configuration()
    conf.set("fs.defaultFS", "hdfs://doitedu01:8020")
    val fs = FileSystem.get(conf)
    val file = new Path("/ip2region/ip2region.db")
    val len = fs.getFileStatus(file).getLen
    val inputStream = fs.open(file)
    val ba = new Array[Byte](len.toInt)
    inputStream.readFully(ba)

    val bc2 = spark.sparkContext.broadcast(ba)


    /**
     * 地理位置解析
     */
    val regionedRdd = washedEventLog.mapPartitions(iter => {
      // 从广播变量中取出geo字典
      val geoDict: collection.Map[String, (String, String, String)] = bc.value
      // 从广播变量中取出ip位置库字典
      val ipDictBytes: Array[Byte] = bc2.value

      // 构造ip段索引搜索器
      val config = new DbConfig()
      val searcher = new DbSearcher(config, ipDictBytes)

      iter.map(
        bean => {
          val longitude = bean.longitude
          val latitude = bean.latitude

          var area: (String, String, String) = (null, null, null)
          // 如果经纬度合法，则用经纬度变换成geohash码，去查询geo地理位置字典
          if (longitude > -180 && longitude < 180 && latitude > -90 && latitude < 90) {
            val geo = GeoHash.geoHashStringWithCharacterPrecision(latitude, longitude, 5)
            area = geoDict.getOrElse(geo, (null, null, null))
          }

          // 判断geo得到的地理位置是否有效，如果无效，则用ip地址再试一次
          if (area._1 == null && StringUtils.isNotBlank(bean.ip)) {
            val block = searcher.memorySearch(bean.ip)
            val arr = block.getRegion.split("\\|")
            if (arr.length > 3) {
              area = (arr(2), arr(3), null)
            }

          }
          // 填充地理位置
          bean.province = if (area._1 == null || area._1.equals("0")) null else area._1
          bean.city = if (area._2 == null || area._2.equals("0")) null else area._2
          bean.district = if (area._3 == null || area._3.equals("0")) null else area._3
          // 返回最后结果
          bean
        }
      )
    })

    /**
     * GUID 标识
     */
    // 加载当日的 guid设备绑定表
    val bindTableRdd = BindTableUtil.loadBindTable(spark, s"${cur_dt}")
    val bindMap = bindTableRdd.collectAsMap()
    val bc3 = spark.sparkContext.broadcast(bindMap)
    val guidedRdd = regionedRdd.map(bean => {
      val guidDict = bc3.value

      var guid: String = bean.account
      // 取出bean中的account，如果有值，则guid=account，否则，用deviceid去guidDict字典中查询
      if (StringUtils.isBlank(bean.account)) {
        guid = guidDict.getOrElse(bean.deviceid, null)
      }

      // 填充guid
      bean.guid = guid

      // 返回结果
      bean
    })

    /**
     * 标记新老访客
     * 规则： 数据上的guid如果在前日中的设备绑定表中存在，则为老访客，否则即为新访客
     */
    // 加载前一日的设备绑定表  // {"deviceid":"","lst":[],"guid":""}
    val bindTableRdd2: RDD[(String, String)] = BindTableUtil.loadBindTable(spark, s"${pre_dt}")
    val yesterdayGUIDs = bindTableRdd2.map(_._2).collect().toSet
    val bc4 = spark.sparkContext.broadcast(yesterdayGUIDs)
    val newOldFlagRdd: RDD[EventLogBean] = guidedRdd.map(bean => {
      val oldGuids = bc4.value
      if (oldGuids.contains(bean.guid)) bean.isNew = 0

      bean
    })


    /**
     * 保存结果
     * 目的地：数仓的DWD层表： dwd.app_event_dtl
     */
    val resultDF = newOldFlagRdd.toDF().withColumnRenamed("timestamp", "ts")

    resultDF.createTempView("res")
    spark.sql(
      s"""
        |
        |insert into table dwd.app_event_dtl partition(dt='${cur_dt}')
        |select
        |account             ,
        |appid               ,
        |appversion          ,
        |carrier             ,
        |deviceid            ,
        |devicetype          ,
        |eventid             ,
        |ip                  ,
        |latitude            ,
        |longitude           ,
        |nettype             ,
        |osname              ,
        |osversion           ,
        |properties          ,
        |releasechannel      ,
        |resolution          ,
        |sessionid           ,
        |ts                  ,
        |province            ,
        |city                ,
        |district            ,
        |guid                ,
        |isNew
        |
        |from res
        |
        |""".stripMargin)


    spark.close()
  }
}
